#include "execution/executor.h"  
#include <iostream>  
#include <sstream>  
#include "execution/catalog.h"  
#include "storage/buffer_pool.h"  
#include <ranges>
#include <algorithm>

Executor::Executor(std::unique_ptr<BufferPool>& bufferPool, std::shared_ptr<Catalog> catalog)
    : bufferPool(bufferPool), catalog(catalog), evaluator(*catalog) {
}  

ExecutionResult Executor::execute(const std::unique_ptr<Execution::PlanNode>& plan) {  
    if (dynamic_cast<Execution::CreateTablePlan*>(plan.get())) {  
        return executeCreateTable(*static_cast<Execution::CreateTablePlan*>(plan.get()));  
    }  
    else if (dynamic_cast<Execution::InsertPlan*>(plan.get())) {  
        return executeInsert(*static_cast<Execution::InsertPlan*>(plan.get()));  
    }  
    else if (dynamic_cast<Execution::SelectPlan*>(plan.get())) {  
        return executeSelect(*static_cast<Execution::SelectPlan*>(plan.get()));  
    }  
    else if (dynamic_cast<Execution::DeletePlan*>(plan.get())) {  
        return executeDelete(*static_cast<Execution::DeletePlan*>(plan.get()));  
    }  
    // Add handling for UpdatePlan (moved to execute method)
    else if (dynamic_cast<Execution::UpdatePlan*>(plan.get())) {
        return executeUpdate(*static_cast<Execution::UpdatePlan*>(plan.get()));
    }

    return { false, "Unknown plan type", {} };  
}  

ExecutionResult Executor::executeCreateTable(const Execution::CreateTablePlan& plan) {  
    TableInfo tableInfo;  
    tableInfo.name = plan.tableName;  
    tableInfo.firstPageId = 0; 

    for (const auto& col : plan.columns) {  
        ColumnInfo colInfo;  
        colInfo.name = col.first;  

        // normalize type to uppercase
        std::string typ = col.second;
        std::transform(typ.begin(), typ.end(), typ.begin(), ::toupper);
        colInfo.type = typ;
        colInfo.length = 0;
        if (typ == "INT") {
            colInfo.length = sizeof(int);
           
        }
        else if (typ == "VARCHAR" || typ == "STRING") {
            colInfo.length = 255;
        }
        tableInfo.columns.push_back(colInfo);  
    }  

    if (catalog->createTable(tableInfo)) {  
        return { true, "Table created successfully", {} };  
    }  
    else {  
        return { false, "Table already exists", {} };  
    }  
}  

ExecutionResult Executor::executeInsert(const Execution::InsertPlan& plan) {  
    TableInfo* tableInfo = catalog->getTable(plan.tableName);  
    if (tableInfo == nullptr) {  
        return { false, "Table '" + plan.tableName + "' does not exist", {} };  
    }  

 
    std::vector<std::string> values = plan.values;  


    if (tableInfo->firstPageId == 0) {  
        Page* firstPage = bufferPool->newPage();  
        if (!firstPage) {  
            return { false, "Failed to allocate first page", {} };  
        }  
        tableInfo->firstPageId = firstPage->getPageId();  
 
        catalog->saveToStorage();
    }  


    uint32_t currentPageId = tableInfo->firstPageId;  
    Page* currentPage = nullptr;  
    
    while (true) {  
        currentPage = bufferPool->getPage(currentPageId);  

        if (currentPage->insertRecord(values, *tableInfo)) {  
            bufferPool->flushPage(currentPageId);  
            catalog->saveToStorage();
            return { true, "Insert executed successfully", {} };  
        }  
 
        PageHeader header = currentPage->getHeader();  
        if (header.nextPageId == 0) {  
            Page* newPage = bufferPool->newPage();  
            if (!newPage) {  
                return { false, "Failed to allocate new page", {} };  
            }  
            
            header.nextPageId = newPage->getPageId();  
            currentPage->setHeader(header);  
            bufferPool->flushPage(currentPageId);  
            
            if (newPage->insertRecord(values, *tableInfo)) {  
                bufferPool->flushPage(newPage->getPageId());  
                return { true, "Insert executed successfully", {} };  
            } else {  
                return { false, "Failed to insert record even after creating new page", {} };  
            }  
        }  

        currentPageId = header.nextPageId;  
    }  
}  

ExecutionResult Executor::executeSelect(const Execution::SelectPlan& plan) {  
    ExecutionResult result;  
    result.success = true;  
    result.message = "Select executed successfully";  
    
    TableInfo* tableInfo = catalog->getTable(plan.tableName);  
    if (tableInfo == nullptr) {  
        return { false, "Table '" + plan.tableName + "' does not exist", {} };  
    }  

    // If table has no data pages, return empty result set
    if (tableInfo->firstPageId == 0) {  
        return result;  
    }  

    // Iterate through all data pages
    uint32_t currentPageId = tableInfo->firstPageId;  
    
    while (currentPageId != 0) {  
        Page* page = bufferPool->getPage(currentPageId);  
        
        // Get all records on the current page
        auto pageRecords = page->getRecords(*tableInfo);  
        for (const auto& record : pageRecords) {  
            // If there's a filter condition, filter the records
            if (plan.filter.empty()) {
                // No filter condition, add all records directly
                result.rows.push_back(record);
            } else {
                // Evaluate filter condition using ExpressionEvaluator
                if (evaluator.evaluate(plan.filter, record, *tableInfo)) {
                    result.rows.push_back(record);
                }
            }  
        }  
        
        // Move to next page
        PageHeader header = page->getHeader();  
        currentPageId = header.nextPageId;  
    }  
    
    return result;  
}  

ExecutionResult Executor::executeDelete(const Execution::DeletePlan& plan) {  
    TableInfo* tableInfo = catalog->getTable(plan.tableName);  
    if (tableInfo == nullptr) {  
        return { false, "Table '" + plan.tableName + "' does not exist", {} };  
    }  

    // If table has no data pages, return success directly
    if (tableInfo->firstPageId == 0) {  
        return { true, "Delete executed successfully", {} };  
    }  

    // Iterate through all data pages to perform deletion
    uint32_t currentPageId = tableInfo->firstPageId;  
    uint32_t deletedCount = 0;  
    
    while (currentPageId != 0) {  
        Page* page = bufferPool->getPage(currentPageId);  
        PageHeader header = page->getHeader();  
        
        // If there's no filter condition, empty the entire table
            if (plan.filter.empty()) {  
            // Create a new empty page
                uint32_t newPageId = 0;  
            Page* newPage = bufferPool->newPage();  
            if (newPage) {  
                newPageId = newPage->getPageId();  
                
                // Update the firstPageId of the table
                tableInfo->firstPageId = newPageId;  
                
                // Keep the new page, other pages can be released through BufferPool's LRU replacement strategy
                bufferPool->flushPage(newPageId);  
            }  
            deletedCount = header.recordCount;  
            break;  
        } else {  
            // Get all records on the current page
            auto pageRecords = page->getRecords(*tableInfo);
            bool recordsDeleted = false;
            
            // Iterate records from back to front to avoid deletion order issues
            for (int i = pageRecords.size() - 1; i >= 0; i--) {
                if (evaluator.evaluate(plan.filter, pageRecords[i], *tableInfo)) {
                    // Delete records that meet the conditions
                    if (page->deleteRecord(i)) {
                        deletedCount++;
                        recordsDeleted = true;
                    }
                }
            }
            
            // If records were deleted, flush the page to disk
            if (recordsDeleted) {
                bufferPool->flushPage(currentPageId);
            }  
        }  
        
        // Move to next page
            currentPageId = header.nextPageId;  
    }  
    
    return { true, "Delete executed successfully, " + std::to_string(deletedCount) + " records deleted", {} };  
}

// Keep the implementation of executeUpdate method (but remove incorrect comment lines)
ExecutionResult Executor::executeUpdate(const Execution::UpdatePlan& plan) {
    TableInfo* tableInfo = catalog->getTable(plan.tableName);
    if (tableInfo == nullptr) {
        return { false, "Table '" + plan.tableName + "' does not exist", {} };
    }

    // If table has no data pages, return success directly
    if (tableInfo->firstPageId == 0) {
        return { true, "Update executed successfully, 0 records updated", {} };
    }

    // Iterate through all data pages to perform updates
    uint32_t currentPageId = tableInfo->firstPageId;
    uint32_t updatedCount = 0;
    
    while (currentPageId != 0) {
        Page* page = bufferPool->getPage(currentPageId);
        PageHeader header = page->getHeader();
        
        // Get all records on the current page
        auto pageRecords = page->getRecords(*tableInfo);
        std::vector<std::vector<std::string>> updatedRecords;
        bool recordsUpdated = false;
        
        // Iterate through records
        for (size_t i = 0; i < pageRecords.size(); ++i) {
            const auto& record = pageRecords[i];
            
            // If there's a filter condition, check if the record meets the condition
            if (plan.filter.empty() || evaluator.evaluate(plan.filter, record, *tableInfo)) {
                // Create new record values
                std::vector<std::string> newRecord = record;
                
                // Apply updates
                for (const auto& [colName, newValue] : plan.columnValues) {
                    // Find column index
                    for (size_t j = 0; j < tableInfo->columns.size(); ++j) {
                        if (tableInfo->columns[j].name == colName) {
                            // Update column value
                            newRecord[j] = newValue;
                            break;
                        }
                    }
                }
                
                // Update record
                    if (page->updateRecord(static_cast<uint16_t>(i), newRecord, *tableInfo)) {
                    updatedCount++;
                    recordsUpdated = true;
                }
            }
        }
        
        // If records were updated, flush the page to disk
            if (recordsUpdated) {
            bufferPool->flushPage(currentPageId);
        }
        
        // Move to next page
        currentPageId = header.nextPageId;
    }
    
    // Save catalog information
    catalog->saveToStorage();
    
    return { true, "Update executed successfully, " + std::to_string(updatedCount) + " records updated", {} };
}